package com.kai.service.mqtt.client;

import com.kai.service.mqtt.handler.MessageHandler;
import com.kai.service.util.ExecutorUtil;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;

@Slf4j
@Component
public class DefaultMqttClient {

    @Value("${mqtt.server-url:}")
    private String serverUrl;

    @Value("${mqtt.client-id:}")
    private String clientId;

    @Value("${mqtt.username:}")
    private String username;

    @Value("${mqtt.password:}")
    private String password;

    @Value("${mqtt.subscribe-topics:}")
    private String subscribeTopics;

    private MqttClient mqttClient;

    private Boolean reconnectRunning = false;

    @PostConstruct
    public void initClient() {
        try {
            mqttClient = new MqttClient(serverUrl, clientId + System.nanoTime(), new MemoryPersistence());
            mqttClient.setCallback(new MqttCallbackExtended() {
                @Override
                public void connectComplete(boolean reconnect, String serverURI) {
                    ExecutorUtil.submit(() -> {
                        try {
                            String[] topics = subscribeTopics.split(",");
                            mqttClient.subscribe(topics, new int[topics.length]);
                        } catch (Exception e) {
                            log.error(e.getMessage(), e);
                        }
                    });
                    log.info("client connect success, reconnect: {}, serverURI: {}", reconnect, serverURI);
                }

                @Override
                public void connectionLost(Throwable throwable) {
                    log.error("client connect fail", throwable);
                    tryReconnect();
                }

                @Override
                public void messageArrived(String topic, MqttMessage mqttMessage) {
                    new MessageHandler(topic, mqttMessage).handle();
                }

                @Override
                public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
                    log.info("send msg succeed topic is : " + iMqttDeliveryToken.getTopics()[0]);
                }
            });
            mqttClient.connect(getMqttConnectOptions());
        } catch (Exception e) {
            log.error("消费者客户端初始化失败", e);
        }
    }

    public void sendMessage(String topic, MqttMessage mqttMessage) {
        ExecutorUtil.execute(() -> {
            try {
                mqttClient.publish(topic, mqttMessage);
            } catch (Exception e) {
                log.error("Send MQTT Msg Failure, Topic: {}, Reason: {}", topic, e);
            }
        });
    }

    private void tryReconnect() {
        if (reconnectRunning) {
            return;
        }
        reconnectRunning = true;
        while (true) {
            log.warn("MQTT trying to reconnect...");
            if (mqttClient.isConnected()) {
                break;
            } else {
                try {
                    mqttClient.connect(getMqttConnectOptions());
                    mqttClient.subscribe(subscribeTopics.split(","));
                    log.warn("MQTT Reconnected successfully..., {}", mqttClient.getServerURI());
                    if (mqttClient.isConnected()) {
                        break;
                    }
                } catch (MqttException ex) {
                    log.error("MQTT Reconnect Exception, Reason Code:{}", ex.getReasonCode(), ex);
                }
                try {
                    Thread.sleep(60000);
                } catch (InterruptedException ex) {
                    log.error("Exception, ", ex);
                }
            }
        }
        reconnectRunning = false;
    }

    /**
     * 获取mqtt连接条件
     *
     * @return
     */
    private MqttConnectOptions getMqttConnectOptions() {
        MqttConnectOptions mqttConnectOptions = new MqttConnectOptions();
        mqttConnectOptions.setUserName(username);
        mqttConnectOptions.setPassword(password.toCharArray());
        mqttConnectOptions.setConnectionTimeout(10);
        mqttConnectOptions.setKeepAliveInterval(30);
        mqttConnectOptions.setCleanSession(true);
        mqttConnectOptions.setAutomaticReconnect(true);

        return mqttConnectOptions;
    }

}
